package chapter2;

import Config.GlobalKafkaConfig;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.TimeUnit;


public class KafkaProducerAnalysis {
    public static final String brokerList = GlobalKafkaConfig.brokerList;
    public static final String topic = GlobalKafkaConfig.topic;


    public static Properties initConfig() {
        Properties props = new Properties();
        props.put("bootstrap.servers", brokerList);
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("client.id", "producer.client.id.demo");
        return props;
    }

    public static Properties initNewConfig() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer.client.id.demo");
        return props;
    }

    public static Properties initPerferConfig() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        return props;
    }

    public static void main(String[] args) throws InterruptedException {
        Properties props = initConfig();
        //添加kafka的拦截器
        props.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,ProducerInterceptorPrefix.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

//        KafkaProducer<String, String> producer = new KafkaProducer<>(props,
//                new StringSerializer(), new StringSerializer());

        for (int i = 0; i < 5; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, "hello, Kafka!");

            try {
                //producer.send(record);
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception exception) {
                        System.out.println("==========================");
                        if (exception == null) {
                            System.out.println("hasOffset:  "+recordMetadata.hasOffset());
                            System.out.println("hasTimestamp:  "+recordMetadata.hasTimestamp());
                            System.out.println("offset:  "+recordMetadata.offset());
                            System.out.println("partition:  "+recordMetadata.partition());
                            System.out.println("timestamp:  "+recordMetadata.timestamp());
                            System.out.println("topic:  "+recordMetadata.topic());
                            System.out.println(recordMetadata.toString());
                        }
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }



        TimeUnit.SECONDS.sleep(5);
    }
}
